package com.atguigu.day02;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * flatMap的使用
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Integer> stream = env.addSource(new ParallelSourceFunction<Integer>() {
            private Boolean flag = true;
            private Random random = new Random();

            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                while (flag) {
                    sourceContext.collect(random.nextInt(1000));
                    TimeUnit.MILLISECONDS.sleep(500);
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });
        //求和
//        stream
//                .keyBy(x->true)
//                .reduce(Integer::sum)
//                .print();
//        //求平均值
//        stream
//                .map(x-> Tuple2.of(x,1)).returns(Types.TUPLE(Types.INT,Types.INT))
//                .keyBy(x->true)
//                .reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
//                    @Override
//                    public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> t0, Tuple2<Integer, Integer> t1) throws Exception {
//                        return Tuple2.of(t0.f0+t1.f0,t1.f1+t1.f1);
//                    }
//                }).returns(Types.TUPLE(Types.INT,Types.INT))
//                .map(x->(double)x.f0/x.f1)
//                .print();
//        //求最大最小值
        stream
                .map(x->Tuple2.of(x,x)).returns(Types.TUPLE(Types.INT,Types.INT))
                .keyBy(x->true)
                .reduce((x,y)->Tuple2.of(
                        Math.max(x.f0, y.f0),
                        Math.min(x.f1, y.f1)
                )).returns(Types.TUPLE(Types.INT,Types.INT))
                .print();
        env.execute();
    }
}
